//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// update_executor.cpp
//
// Identification: src/execution/update_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//
#include <memory>

#include "execution/executors/update_executor.h"

namespace bustub {

UpdateExecutor::UpdateExecutor(ExecutorContext *exec_ctx, const UpdatePlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx), plan_(plan), child_executor_(std::move(child_executor)) {
  // As of Fall 2022, you DON'T need to implement update executor to have perfect score in project 3 / project 4.
}

void UpdateExecutor::Init() {
  child_executor_->Init();
  updated_ = false;
  table_info_ = exec_ctx_->GetCatalog()->GetTable(plan_->table_oid_);
}

auto UpdateExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (updated_) {
    return false;
  }
  RID ch_rid;
  Tuple ch_tuple;
  Catalog *catalog = exec_ctx_->GetCatalog();
  const auto &expr_vec = plan_->target_expressions_;
  int value_cnt = 0;
  std::vector<IndexInfo *> index_infos = catalog->GetTableIndexes(table_info_->name_);
  while (child_executor_->Next(&ch_tuple, &ch_rid)) {
    auto meta = table_info_->table_->GetTupleMeta(ch_rid);
    if (not meta.is_deleted_) {
      meta.is_deleted_ = true;
      table_info_->table_->UpdateTupleMeta(meta, ch_rid);
      for (auto key_info : index_infos) {
        auto index_meta = key_info->index_->GetMetadata();
        key_info->index_->DeleteEntry(
            ch_tuple.KeyFromTuple(table_info_->schema_, *index_meta->GetKeySchema(), index_meta->GetKeyAttrs()), ch_rid,
            nullptr);
      }

      meta = TupleMeta{INVALID_TXN_ID, INVALID_TXN_ID, false};
    }

    std::vector<Value> vals;
    vals.reserve(expr_vec.size());
    for (const auto &expr_p : expr_vec) {
      vals.emplace_back(expr_p->Evaluate(&ch_tuple, table_info_->schema_));
    }
    Tuple new_tuple(vals, &table_info_->schema_);
    RID inserted_rid = table_info_->table_->InsertTuple(meta, new_tuple).value();
    for (auto key_info : index_infos) {
      auto index_meta = key_info->index_->GetMetadata();
      key_info->index_->InsertEntry(
          new_tuple.KeyFromTuple(table_info_->schema_, *index_meta->GetKeySchema(), index_meta->GetKeyAttrs()),
          inserted_rid, nullptr);
    }
    ++value_cnt;
  }

  *tuple = Tuple(std::vector<Value>(GetOutputSchema().GetColumnCount(), Value(TypeId::INTEGER, value_cnt)),
                 &GetOutputSchema());

  updated_ = true;
  return true;
}
}  // namespace bustub
